import mqtt from 'mqtt';
import { bus } from './bus';
import { getAllGateways } from './proc/persist/gateway';
import { DataSource, Repository } from 'typeorm';
import { Gateway } from './entity/gateway.entity';
import { envConfig } from './config/index'

let clients: Record<string, mqtt.MqttClient> = {};
// 存储各主机的心跳定时器
let heartbeatIntervals: Record<string, NodeJS.Timeout> = {};

export async function init(host: string, AppDataSource?: DataSource): Promise<mqtt.MqttClient> {
    if (clients[host]) {
        console.log(`MQTT 客户端已连接到 ${host}`);
        return clients[host];
    }

    const client = mqtt.connect(host);

    client.on('connect', () => {
        console.log(`已连接到 MQTT 主机: ${host}`);
    });

    client.on('error', (err) => {
        console.error(`MQTT 连接错误 (${host}):`, err);
    });

    clients[host] = client;
    if (AppDataSource) {
        connect(host, AppDataSource);
    } else {
        connect(host, null);
    }
    if (host == envConfig.MQTT_SUZHOU_HOST) {
        // 初始化该客户端的心跳
        setupHeartbeat(host);
    }

    return client;
}

function connect(brokerUrl: string, AppDataSource?: DataSource | null) {
    let gatewayRepo: any;
    if (AppDataSource) {
        gatewayRepo = AppDataSource.getRepository(Gateway);
    }
    let client: mqtt.MqttClient | null = null
    if (!brokerUrl) {
        throw new Error('Broker URL is not set')
    }

    client = mqtt.connect(brokerUrl, {
        keepalive: 60,
        reconnectPeriod: 1000, // 重连间隔时间(毫秒)
        connectTimeout: 30 * 1000,
        clean: true
    })

    client.on('connect', async () => {
        console.log('mqtt connect')
        if (!client) {
            console.error('MQTT client not valid');
            return;
        }
        console.log('MQTT connected to broker:', brokerUrl)

        client.subscribe('wb/report', (err) => {
            // console.error('Error subscribing:', 'wb/report')
        })
        client.subscribe('wb/report/msgpack', (err) => {
            // console.error('Error subscribing:', 'wb/report/msgpack')
        })
        // 上报数据及升级状态
        client.subscribe('wristbrave/adv/report', (err) => {
            // console.error('Error subscribing:', 'wb/report/msgpack')
        })
        // 上报升级/写名字等 指定网关
        client.subscribe('gw/action', (err) => {
            // console.error('Error subscribing:', 'wb/report/msgpack')
        })
        if (AppDataSource) {
            // 从数据库中的网关表获取网关地址, 并订阅对应的topic
            subscribeGateway(client, gatewayRepo)
        }



    })

    client.on('message', (topic: string, message: Buffer, packet) => {
        // console.log('MQTT message received:', topic, message.length, packet)
        if (topic === 'wb/report/msgpack') {
            bus.emit('wb.adv.received:msgpack', message)
        }
        if (topic === 'wristbrave/adv/report') {
            // bus.emit('wb.adv.received', message)
        }
        // ota
        // 使用正则表达式匹配网关报告主题
        const gwReportTopicRegex = /^gw\/report\/[0-9a-fA-F]{2}(:[0-9a-fA-F]{2}){5}$/;
        if (gwReportTopicRegex.test(topic)) {
            console.log('topic1', topic)
            bus.emit('gw.report', JSON.parse(message.toString()))
        }

    })

    client.on('error', (error) => {
        console.error('MQTT connection error:', error);
    })

    client.on('disconnect', (error) => {
        console.error('MQTT connection error:', error);
    })

    // 订阅对应的 topic (mac)
    async function subscribeGateway(client: mqtt.MqttClient, gatewayRepo: Repository<Gateway>) {
        // 从数据库获取网关列表
        try {
            const gateways = await getAllGateways(gatewayRepo);

            // 日志显示网关数量
            console.log(`共找到 ${gateways.length} 个待订阅网关`);

            // 使用 Set 存储已处理MAC地址（去重）
            const uniqueMacs = new Set<string>();

            // 并行执行所有订阅
            const subscriptionResults = await Promise.allSettled(
                gateways.map(item => {
                    // 检查是否重复MAC
                    if (uniqueMacs.has(item.mac)) {
                        console.warn(`发现重复MAC地址: ${item.mac}，跳过重复订阅`);
                        return Promise.resolve();
                    }

                    // 记录已处理MAC
                    uniqueMacs.add(item.mac);

                    // 返回订阅Promise
                    return new Promise<void>((resolve, reject) => {
                        client.subscribe(`gw/report/${item.mac}`, (err) => {
                            if (err) {
                                console.error(`订阅 gw/report/${item.mac} 失败:`, err);
                                reject(err);
                            } else {
                                console.log(`成功订阅 gw/report/${item.mac}`);
                                resolve();
                            }
                        });
                    });
                })
            );

            // 统计订阅结果
            const 成功数 = subscriptionResults.filter(r => r.status === 'fulfilled').length;
            const 失败数 = subscriptionResults.filter(r => r.status === 'rejected').length;

            console.log(`订阅结果: 成功 ${成功数} 个，失败 ${失败数} 个`);

        } catch (error) {
            // console.error('初始化网关订阅失败:', error);
            // 可根据实际需求决定是否抛出错误
        }
    }
}


/**
 * 设置心跳定时器
 * @param host MQTT主机地址
 * @param interval 心跳间隔时间(毫秒)，默认 1分钟
 */
function setupHeartbeat(host: string, interval: number = 60 * 1000) {
    // 清理现有的心跳定时器
    cleanupHeartbeat(host);

    // 设置新的定时器
    heartbeatIntervals[host] = setInterval(() => {
        const client = clients[host];
        if (client && client.connected) {
            const heartbeatMessage = JSON.stringify({
                action: 'heartbeat'
            });

            // 发布心跳消息到指定主题
            client.publish('gw/action', heartbeatMessage, {
                qos: 1 // 确保消息至少送达一次
            }, (err) => {
                if (err) {
                    console.error(`向 ${host} 发送心跳失败:`, err);
                } else {
                    // console.log(`心跳已发送至 ${host}，时间: ${new Date().toISOString()}`);
                }
            });
        }
    }, interval);
}

/**
 * 清理心跳定时器
 * @param host 要清理的MQTT主机地址
 */
function cleanupHeartbeat(host: string) {
    if (heartbeatIntervals[host]) {
        clearInterval(heartbeatIntervals[host]);
        delete heartbeatIntervals[host];
    }
}

